package javaVersion.learn.streamProcess;

import javaVersion.pojo.Pro6_Stream_WordWithCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @program: myFlink
 * @description: 读取Socket数据源, 单词计数
 * @author: WincoMa
 * @create: 2020-07-20 09:28
 **/
public class Pro6_SocketWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        DataStreamSource<String> socketTextStream = senv.socketTextStream("hadoop1", 9999);
        //将统计结果封装到Tuple里
        wordCountWithTuple(socketTextStream,senv);
        //将统计结果封装到POJO里
        wordCountWithPojo(socketTextStream,senv);

    }

    private static void wordCountWithPojo(DataStreamSource<String> socketTextStream,StreamExecutionEnvironment senv) throws Exception {
        SingleOutputStreamOperator<Pro6_Stream_WordWithCount> windowCount = socketTextStream.flatMap(new FlatMapFunction<String, Pro6_Stream_WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<Pro6_Stream_WordWithCount> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(new Pro6_Stream_WordWithCount(word, 1L));
                }
            }
        })
                .keyBy("word")
                .timeWindow(Time.seconds(2), Time.seconds(1))
                .sum("count");
        windowCount.print();
        senv.execute();

    }

    private static void wordCountWithTuple(DataStreamSource<String> socketTextStream, StreamExecutionEnvironment senv) {
        socketTextStream.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                //此种方式是每次数据传入都是一个元素,不需要进行任何处理
//                out.collect(new Tuple2<>(value,1));
                //此种方式是一次输入一行数据,每次输入的格式是数据加空格,需要进行切分
                String[] words = value.split(" ");
                //由于一次可能多个单词,需要进行遍历
                for (String word : words) {
                    //将所有获取的单词分装到Tuple中,并且计数为1
                    out.collect(new Tuple2<>(word,1));
                }
            }
        })
                .keyBy(0) //对Tuple中的第一个字段(word)分组统计
                .sum(1) //对第二个字段(1)进行累加统计
                .print();
    }
}
